package twitterhadoop.app;

import java.io.IOException;
import java.net.URI;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Task;
import org.apache.hadoop.mapred.Task.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import twitterhadoop.hadoop.io.TextArrayWritable;
import twitterhadoop.hadoop.io.TextFileTwitterInputFormat;
import twitterhadoop.hadoop.io.TwitterInputFormat;
import cache.io.CacheInputFormat;

import com.sampullara.cli.Args;

public class TwitterClosedItemSet extends Configured implements Tool {
	private static final Log LOG = LogFactory.getLog(TwitterClosedItemSet.class);
	
	public static final String MIN_SUPP_PARAMETER_NAME = "minSup";
	public static final String JOB_NAME = "Mine Closed Itemsets on Tweets";
	
	public static final String CLOSED_ITEMSETS_PATH_ATTR = "closed.itemsets.path";
	public static final String CLOSED_ITEMSETS_PATH = "closed.itemsets";
	public static final String JOB_COUNTER = "job.counter";
	
	private long totalCommunicationCost = 0;
	private long totalNumberOfClosedItemsets = 0;
	
	/*
	public static final String RESULT_STRING = "result_closed_itemset";
	public static final Text   RESULT_TEXT = new Text(RESULT_STRING);
	*/
	
	public static final String CLOSED_ITEMSETS_SEPERATOR = ":";
	
	private int countNumOfTweets(CommandLineArgs cla) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf);

		job.setJarByClass(CountingMapper.class);
		job.setMapperClass(CountingMapper.class);
		job.setJobName("Count the number of tweets");
		
		setInput(job, cla);
		
		Path outputPath = new Path("/tmp/1.1");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}		
		FileOutputFormat.setOutputPath(job, outputPath);

		job.waitForCompletion(true);
		return (int) job.getCounters().findCounter(Task.Counter.MAP_INPUT_RECORDS).getValue();
	}
	
	private void setInput(Job job, CommandLineArgs cla) throws IOException {

		if (cla.inputFileName != null) {
			TextFileTwitterInputFormat.addInputPath(job, new Path(cla.inputFileName));
			job.setInputFormatClass(TextFileTwitterInputFormat.class);
			return;
		}

		if (cla.useCache) {			
			// Twitter input with cache
			job.setInputFormatClass(CacheInputFormat.class);
			CacheInputFormat.setDelegateInputFormatData(
					job, 
					TwitterInputFormat.class, 
					String.format("%s:%s:%d", cla.query, cla.qDate, cla.days));			
		} else {
			// Twitter input no cache
			job.setInputFormatClass(TwitterInputFormat.class);			
		}
		
		TwitterInputFormat.setQuery(job, cla.query, cla.qDate, cla.days);
		TwitterInputFormat.setLimit(job, cla.tweetsLimit);
		
	}
	
	public int run(String[] args) throws Exception {
		// General stuff
	    CommandLineArgs cla = new CommandLineArgs();
	    try {
	    	Args.parse(cla, args);	    	
	    } catch (IllegalArgumentException e) {
	    	Args.usage(cla);
	    	System.exit(-1);
	    }
	    	    
		Path outputPath = new Path(cla.outputPathName);
		FileSystem fs = FileSystem.get(this.getConf());
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}		
		
		int numOfItemsets = countNumOfTweets(cla);
		double minSupport = numOfItemsets * cla.minSupport;
		int suppurtValue = (int) Math.ceil(minSupport);
		LOG.info(String.format("Number of itemsets it %d. Minimum items for support is %d", numOfItemsets, suppurtValue));
		boolean endProgram = false;
		int jobCounter = 1;
		long currNumOfClosedItemsets = 0;
		
		
		while (!endProgram) {
			// Configuration
			Job job = new Job(getConf());
			Configuration conf = job.getConfiguration();				
			conf.setInt(MIN_SUPP_PARAMETER_NAME, suppurtValue);
			conf.setInt(JOB_COUNTER, jobCounter);		
					
			job.setJarByClass(TwitterClosedItemSet.class);
			job.setJobName(String.format("%s job number %d", JOB_NAME, jobCounter));
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			//job.setMapOutputKeyClass(TextArrayWritable.class);
			job.setMapOutputKeyClass(ClosedItemSet.class);
			job.setMapOutputValueClass(TextArrayWritable.class);
			job.setMapperClass(Map.class);
			job.setReducerClass(Reduce.class);
			
			
			// Sets the input according to the command line flag useCache.
			setInput(job, cla);

			// String closedItemsetsFilename = cla.outputPathName + "/" + TwitterClosedItemSet.CLOSED_ITEMSETS_PATH + "." + (jobCounter-1);
			String closedItemsetsFilename = String.format("%s/%s.%s/", cla.outputPathName, TwitterClosedItemSet.CLOSED_ITEMSETS_PATH, (jobCounter-1));
			Path closedItemsetsDirectory = new Path(closedItemsetsFilename);
			fs = FileSystem.get(conf);
			if (fs.exists(closedItemsetsDirectory)) {
				FileStatus[] status = fs.listStatus(closedItemsetsDirectory);
				for (FileStatus fileStatus : status) {
					String uriStr = fileStatus.getPath().toString();
					DistributedCache.addCacheFile(new URI(uriStr), conf);	
				}
				
				
			}		

			/*
			String closedItemsetsFilename = cla.outputPathName + "/" + TwitterClosedItemSet.CLOSED_ITEMSETS_PATH + "." + (jobCounter-1);
			Path closedItemsetsFile = new Path(closedItemsetsFilename);
			fs = FileSystem.get(conf);
			if (fs.exists(closedItemsetsFile)) {
				DistributedCache.addCacheFile(new URI(closedItemsetsFilename), conf);
			}
			*/		
			
			// job.setOutputFormatClass(SequenceFileOutputFormat.class);
			job.setOutputFormatClass(TextOutputFormat.class);
			String outputDirName = cla.outputPathName + "/out" + "." + jobCounter;
			outputPath = new Path(outputDirName);
			
			if (fs.exists(outputPath)) {
				fs.delete(outputPath, true);
			}		
			
			FileOutputFormat.setOutputPath(job, outputPath);

			//job.waitForCompletion(true);

			if(!job.waitForCompletion(true)) {
				System.err.println("Something happened.");
			}
			
			currNumOfClosedItemsets = job.getCounters().findCounter(InternalCounters.REDUCER_OUTPUT).getValue();
			LOG.info(String.format("Iteration number: %d, number of closed itemsets: %d", jobCounter, currNumOfClosedItemsets));
			
			this.totalCommunicationCost += job.getCounters().findCounter(Counter.MAP_INPUT_RECORDS).getValue();
			this.totalCommunicationCost += job.getCounters().findCounter(Counter.REDUCE_INPUT_RECORDS).getValue();				
			this.totalNumberOfClosedItemsets += currNumOfClosedItemsets;
			
			if (currNumOfClosedItemsets == 0) {
				endProgram = true;
			} else {
				jobCounter++;
			}

			
		}
		
		LOG.info("Total communication cost: " + this.totalCommunicationCost);
		LOG.info(String.format("Total number of closed itemsets: %d", this.totalNumberOfClosedItemsets));
		return 0;
	}

	public static void main(String[] args) throws Exception {
		long startTime = System.currentTimeMillis();
		int ret = ToolRunner.run(new TwitterClosedItemSet(), args);
		long stopTime = System.currentTimeMillis();
		LOG.info("Running time in ms: " + (stopTime - startTime));
		System.exit(ret);
	}

}